package cn.whuc.spark.operator

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo_combineByKey {
  def main(args: Array[String]): Unit = {
    // 1 创建sparkContext
    val sc: SparkContext = new SparkContext(
      new SparkConf()
        .setMaster("local[*]")
        .setAppName(" ")
    )

    // 2 编写代码
    val rdd1: RDD[(String, Int)] = sc.makeRDD(List(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98)), 2)

    val rdd2: RDD[(String, (Int, Int))] = rdd1.combineByKey(
      // acc的结构，结合value的变化
      num => (num, 1),
      // 分区内 累加器中 值相加 次数累加
      (acc: (Int, Int), value: Int) => (acc._1 + value, acc._2 + 1),
      // 分区间 不同累加器之间的 值 累加
      (acc1: (Int, Int), acc2: (Int, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2)
      // 此时原来RDD的结构变了
    )

    val rdd3: RDD[(String, Double)] = rdd2.mapValues(t => if (t._2 > 0) t._1 / t._2.toDouble else 0.0)


    rdd3.collect().foreach(println)

    // 3 关闭上下文对象
    sc.stop()
  }
}
